package local_message

import (
	"context"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
	"go-demo-2025/common"
	"go-demo-2025/utils/kafkaUtil"
	"time"
)

// kafka生产者: 本地消息表方案：本地消息表通过数据库事务将业务操作和消息记录绑定，确保两者要么同时成功，要么同时失败。

// LocalMessage 本地消息表模型
type LocalMessage struct {
	ID            int64     `json:"id"`
	BizType       string    `json:"biz_type"`
	BizID         string    `json:"biz_id"`
	Topic         string    `json:"topic"`
	Content       string    `json:"content"`
	Status        int       `json:"status"`
	RetryCount    int       `json:"retry_count"`
	NextRetryTime time.Time `json:"next_retry_time"`
	CreateTime    time.Time `json:"create_time"`
	UpdateTime    time.Time `json:"update_time"`
}

// 业务操作+消息记录（原子事务）
func CreateOrderAndRecordMessage(ctx context.Context, orderID string, amount int64) error {
	// 开启数据库事务
	tx, err := common.SqlDB.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("开启事务失败: %v", err)
	}
	defer func() {
		if r := recover(); r != nil {
			tx.Rollback()
		}
	}()

	// 1. 执行业务操作（创建订单）
	orderSQL := `INSERT INTO orders (order_id, amount, status, create_time) 
	             VALUES (?, ?, 'pending', NOW())`
	_, err = tx.ExecContext(ctx, orderSQL, orderID, amount)
	if err != nil {
		tx.Rollback()
		return fmt.Errorf("创建订单失败: %v", err)
	}

	// 2. 记录本地消息
	msgContent := fmt.Sprintf(`{"order_id":"%s","amount":%d,"status":"pending"}`, orderID, amount)
	nextRetryTime := time.Now().Add(1 * time.Minute) // 首次重试时间
	msgSQL := `INSERT INTO local_message 
	           (biz_type, biz_id, topic, content, status, next_retry_time)
	           VALUES (?, ?, ?, ?, 0, ?)`
	_, err = tx.ExecContext(ctx, msgSQL, "order_create", orderID, "order_events", msgContent, nextRetryTime)
	if err != nil {
		tx.Rollback()
		return fmt.Errorf("记录本地消息失败: %v", err)
	}

	// 提交事务
	return tx.Commit()
}

// 消息重试定时任务（单独协程运行）
// 使用命令查看当前kafka中的数据：kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic order_events --from-beginning
func StartMessageRetryTask(ctx context.Context) {
	ticker := time.NewTicker(30 * time.Second) // 每30秒扫描一次
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			fmt.Println("重试任务退出")
			return
		case <-ticker.C:
			ProcessRetryMessages(ctx)
		}
	}
}

// 处理待重试消息
func ProcessRetryMessages(ctx context.Context) {
	// 1. 查询待重试的消息（状态0且下次重试时间已到）
	querySQL := `SELECT id, biz_type, biz_id, topic, content, retry_count 
	             FROM local_message 
	             WHERE status = 0 AND next_retry_time <= NOW()
	             LIMIT 100` // 分批处理，避免锁表

	rows, err := common.SqlDB.QueryContext(ctx, querySQL)
	if err != nil {
		fmt.Printf("查询重试消息失败: %v\n", err)
		return
	}
	defer rows.Close()

	// 2. 逐个处理消息
	for rows.Next() {
		var msg LocalMessage
		err := rows.Scan(&msg.ID, &msg.BizType, &msg.BizID, &msg.Topic, &msg.Content, &msg.RetryCount)
		if err != nil {
			fmt.Printf("扫描消息失败: %v\n", err)
			continue
		}

		// 3. 发送消息到Kafka
		err = kafkaUtil.SendKafkaMessage(ctx, msg.Topic, msg.BizID, msg.Content)
		if err == nil {
			// 发送成功：更新状态为已发送
			updateSQL := `UPDATE local_message 
			              SET status = 1, update_time = NOW()
			              WHERE id = ?`
			_, err = common.SqlDB.ExecContext(ctx, updateSQL, msg.ID)
			if err != nil {
				fmt.Printf("消息发送成功,但更新消息状态失败: %v\n", err)
			}
			fmt.Printf("消息发送成功: id=%d, biz_id=%s\n", msg.ID, msg.BizID)
		} else {
			fmt.Println("消息发送失败的原因:", err)
			// 发送失败：更新重试次数和下次重试时间（指数退避）
			newRetryCount := msg.RetryCount + 1
			if newRetryCount >= 5 {
				// 超过最大重试次数，标记为失败
				updateSQL := `UPDATE local_message 
				              SET status = 2, retry_count = ?, update_time = NOW()
				              WHERE id = ?`
				_, err = common.SqlDB.ExecContext(ctx, updateSQL, newRetryCount, msg.ID)
				fmt.Printf("消息发送失败（已达最大重试次数）: id=%d, err=%v\n", msg.ID, err)
			} else {
				// 指数退避：1, 2, 4, 8分钟
				nextRetryTime := time.Now().Add(time.Duration(1<<newRetryCount) * time.Minute)
				updateSQL := `UPDATE local_message 
				              SET retry_count = ?, next_retry_time = ?, update_time = NOW()
				              WHERE id = ?`
				_, err = common.SqlDB.ExecContext(ctx, updateSQL, newRetryCount, nextRetryTime, msg.ID)
				fmt.Printf("消息发送失败（重试中）: id=%d, retry_count=%d, err=%v\n", msg.ID, newRetryCount, err)
			}
		}
	}
}
